package com.bst.datax.executor.service.jobhandler;

import com.bst.common.utils.StringUtils;
import com.bst.common.utils.sql.EtlUtil;
import com.bst.datatx.core.biz.model.JobParam;
import com.bst.datax.executor.service.logparse.LogStatistics;

import java.sql.*;

//@Component
//@StepScope
public class DataBatchInsertHandler {

    private String insertSql;
    private String querySql;
    private final int commitInterval=10000; // 提交间隔

    private JobParam jobParam;
    private LogStatistics logStatistics;

    public DataBatchInsertHandler(JobParam jobParam,LogStatistics statistics) {
        this.jobParam = jobParam;
        this.logStatistics = statistics;
        String dayBegin = jobParam.getDayBegin();
        String dayEnd = jobParam.getDayEnd();
        insertSql = jobParam.getInsertSql();
        querySql = EtlUtil.variReplact(jobParam.getQuerySql(),dayBegin,dayEnd,"sqlserver");
    }

    public void execute(Connection conn) throws Exception {
        Statement stmt = null;
        ResultSet rs = null;
        if(StringUtils.isEmpty(insertSql) || StringUtils.isEmpty(querySql)) {
            return ;
        }
        int columnCount=0,readCount=0,writeCount=0;
        try {
            conn.setAutoCommit(false); // 关闭自动提交
            // 1. 创建源表游标
            stmt = conn.createStatement(
                    ResultSet.TYPE_FORWARD_ONLY,
                    ResultSet.CONCUR_READ_ONLY);
            stmt.setFetchSize(commitInterval);

            rs = stmt.executeQuery(querySql);

            // 2. 准备目标表插入语句
            PreparedStatement pstmt = conn.prepareStatement(insertSql);
            ResultSetMetaData meta = rs.getMetaData();
            columnCount = meta.getColumnCount();
            while (rs.next()) {
                // 绑定参数
                for (int i = 1; i <= columnCount; i++) {
                    pstmt.setObject(i, rs.getObject(i));
                }
                pstmt.addBatch();

                // 定期提交
                if (++readCount % commitInterval == 0) {
                    pstmt.executeBatch();
                    conn.commit();
                    writeCount +=readCount;
                }
            }

            // 提交剩余记录
            if (readCount % commitInterval != 0) {
                pstmt.executeBatch();
                conn.commit();
                writeCount +=readCount;
            }

        } finally {
            // 关闭资源
            if (rs != null) rs.close();
            if (stmt != null) stmt.close();
            if (conn != null) conn.close();
            logStatistics.setTaskRecordReaderNum(readCount);
            logStatistics.setTaskRecordWriteFailNum(readCount-writeCount);
        }
    }

}
